/**
 * Licensed to the Apache Software Foundation (ASF) under one
 * or more contributor license agreements.  See the NOTICE file
 * distributed with this work for additional information
 * regarding copyright ownership.  The ASF licenses this file
 * to you under the Apache License, Version 2.0 (the
 * "License"); you may not use this file except in compliance
 * with the License.  You may obtain a copy of the License at
 *
 *     http://www.apache.org/licenses/LICENSE-2.0
 *
 * Unless required by applicable law or agreed to in writing, software
 * distributed under the License is distributed on an "AS IS" BASIS,
 * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
 * See the License for the specific language governing permissions and
 * limitations under the License.
 */

package org.apache.zookeeper.server;

import java.io.PrintWriter;
import java.io.StringWriter;
import java.util.ArrayList;
import java.util.Collections;
import java.util.Date;
import java.util.HashMap;
import java.util.HashSet;
import java.util.Map.Entry;
import java.util.concurrent.ConcurrentHashMap;

import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.apache.zookeeper.KeeperException;
import org.apache.zookeeper.KeeperException.SessionExpiredException;

/**
 * This is a full featured SessionTracker. It tracks session in grouped by tick
 * interval. It always rounds up the tick interval to provide a sort of grace
 * period. Sessions are thus expired in batches made up of sessions that expire
 * in a given interval.
 */
public class SessionTrackerImpl extends Thread implements SessionTracker {
    private static final Logger LOG = LoggerFactory.getLogger(SessionTrackerImpl.class);

    /** key是sessionId，value是对应的会话 */
    HashMap<Long, SessionImpl> sessionsById = new HashMap<>();
    /** key是某个过期时间，value是会话集合，表示这个过期时间过后就超时的会话集合 */
    HashMap<Long, SessionSet> sessionSets = new HashMap<>();
    /** key是sessionId, value是该会话的超时周期(不是时间点) */
    ConcurrentHashMap<Long, Integer> sessionsWithTimeout;
    /** 下个会话的id */
    long nextSessionId = 0;
    /** 下一次进行超时检测的时间 */
    long nextExpirationTime;
    /** 超时检测的周期，多久检测一次 */
    int expirationInterval;

    public static class SessionImpl implements Session {
        SessionImpl(long sessionId, int timeout, long expireTime) {
            this.sessionId = sessionId;
            this.timeout = timeout;
            this.tickTime = expireTime; // 下一次session可能会过期的时间，
            // 当前是12:03，sessionTimeou=120s, 下次过期时间 12:03 + 2min = 12:05
            isClosing = false;
        }

        /** 会话id，全局唯一 */
        final long sessionId;
        /** 会话超时时间 */
        final int timeout;
        /** 下次会话的超时时间点,会不断刷新 */
        long tickTime;
        /** 是否被关闭,如果关闭则不再处理该会话的新请求 */
        boolean isClosing;

        Object owner;

        @Override
        public long getSessionId() { return sessionId; }
        @Override
        public int getTimeout() { return timeout; }
        @Override
        public boolean isClosing() { return isClosing; }
    }

    /**
     *  高8位就是myid, 低位56位是当前时间戳
     *
     * 比如当前时间转成64位之后
     * 当前时间为2020-08-09 11:20:20 转成毫秒数之后时 1596943220000，转成64位的long类型如下
     * 0000 0000 0000 0000 0000 0001 0111 0011 1101 0001 0011 1011 1011 1101 0010 0000
     * 向左移动24位
     * 0111 0011 1101 0001 0011 1011 1011 1101 0010 0000 0000 0000 0000 0000 0000 0000
     * 然后向右移动8位，此时为 nextSid
     * 0000 0000 0111 0011 1101 0001 0011 1011 1011 1101 0010 0000 0000 0000 0000 0000
     *
     * 接下来是 id << 56
     * id是我们自己设置的myid，一般都是一个比较小的数字，1、2、3、4等等这样的，一个集群有5态机器，一般都是1、2、3、4、5
     * 如果此时id = 5  转成二进制
     * 0000 0000 0000 0000 0000 0000 0000 0000 0000 0000 0000 0000 0000 0000 0000 0101
     * 向左移动56位，得到数据
     * 0000 0101 0000 0000 0000 0000 0000 0000 0000 0000 0000 0000 0000 0000 0000 0000
     *
     * 然后和nextSid做或操作，或操作是 有1得1，没有1得0，即
     * 0000 0000 0111 0011 1101 0001 0011 1011 1011 1101 0010 0000 0000 0000 0000 0000
     * 0000 0101 0000 0000 0000 0000 0000 0000 0000 0000 0000 0000 0000 0000 0000 0000
     * 得到
     * 0000 0101 0111 0011 1101 0001 0011 1011 1011 1101 0010 0000 0000 0000 0000 0000
     *
     * @param id
     * @return
     */
    public static long initializeNextSession(long id) {
        long nextSid = 0;
        // 当前时间戳（64位二进制数据），左移24位，又右移8位
        nextSid = (System.currentTimeMillis() << 24) >> 8;
        // id = myid, 左移56位
        nextSid =  nextSid | (id <<56);
        // 高8位就是myid, 低位56位是当前时间戳
        return nextSid;
    }

    static class SessionSet {
        HashSet<SessionImpl> sessions = new HashSet<>();
    }

    /** 用于server检测client超时之后给client发送 会话关闭的请求 */
    SessionExpirer expirer;

    /**
     * 计算出最近的 一下次统一检测过期的时间
     * @param time
     * @return
     */
    private long roundToInterval(long time) {
        // expirationInterval: 用来后续在后台线程里每隔这么多时间检查session是否超时的，
        // 默认跟ticktime是一样大的，默认2秒钟
        // 默认是 (12:05 / 2s +1) * 2s , 让expiretime过期时间应该是expirationInterval的倍数
        // (2/2 + 1) * 2 = 4
        // (3/2 + 1) * 2 = 4
        // (6/2 + 1) * 2 = 8
        // (7/2 + 1) * 2 = 8
        return (time / expirationInterval + 1) * expirationInterval;
    }

    public SessionTrackerImpl(SessionExpirer expirer, ConcurrentHashMap<Long, Integer> sessionsWithTimeout, int tickTime, long sid){
        super("SessionTracker");
        this.expirer = expirer;
        this.expirationInterval = tickTime;
        this.sessionsWithTimeout = sessionsWithTimeout;
        nextExpirationTime = roundToInterval(System.currentTimeMillis());
        // 计算sessionId
        this.nextSessionId = initializeNextSession(sid);
        for (Entry<Long, Integer> e : sessionsWithTimeout.entrySet()) {
            addSession(e.getKey(), e.getValue());
        }
    }

    /** 超时检测的线程是否在运行 */
    volatile boolean running = true;

    volatile long currentTime;

    @Override
    synchronized public void dumpSessions(PrintWriter pwriter) {
        pwriter.print("Session Sets (");
        pwriter.print(sessionSets.size());
        pwriter.println("):");
        ArrayList<Long> keys = new ArrayList<Long>(sessionSets.keySet());
        Collections.sort(keys);
        for (long time : keys) {
            pwriter.print(sessionSets.get(time).sessions.size());
            pwriter.print(" expire at ");
            pwriter.print(new Date(time));
            pwriter.println(":");
            for (SessionImpl s : sessionSets.get(time).sessions) {
                pwriter.print("\t0x");
                pwriter.println(Long.toHexString(s.sessionId));
            }
        }
    }

    @Override
    synchronized public String toString() {
        StringWriter sw = new StringWriter();
        PrintWriter pwriter = new PrintWriter(sw);
        dumpSessions(pwriter);
        pwriter.flush();
        pwriter.close();
        return sw.toString();
    }

    /**
     * 等到下一次超时检测的周期，把对应的桶中的会话全部标记关闭，给对应client发送 会话关闭的请求
     */
    @Override
    synchronized public void run() {
        try {
            while (running) {
                currentTime = System.currentTimeMillis();
                // 如果下一次超时检测的时间还没到，就等
                if (nextExpirationTime > currentTime) {
                    this.wait(nextExpirationTime - currentTime);
                    continue;
                }
                SessionSet set;
                // 进行会话清理,这个"桶"中的会话都超时了
                set = sessionSets.remove(nextExpirationTime);
                if (set != null) {
                    // 将nextExpirationTime对应的session关闭
                    for (SessionImpl s : set.sessions) {
                        setSessionClosing(s.sessionId); // 标记关闭
                        expirer.expire(s);  // 发起会话关闭请求
                    }
                }
                nextExpirationTime += expirationInterval;   // 设置下一次清理的时间
            }
        } catch (InterruptedException e) {
            LOG.error("Unexpected interruption", e);
        }
        LOG.info("SessionTrackerImpl exited loop!");
    }

    /**
     * 每次接收到客户端的消息之后就进行的操作，其主要是延长对应的session的超时的时间
     * @param sessionId
     * @param timeout
     * @return
     */
    @Override
    synchronized public boolean touchSession(long sessionId, int timeout) {
        SessionImpl s = sessionsById.get(sessionId);
        // 对应session已经被标识为关闭
        if (s == null || s.isClosing()) {
            return false;
        }
        // 计算出新的过期时间
        long expireTime = roundToInterval(System.currentTimeMillis() + timeout);
        if (s.tickTime >= expireTime) {
            return true;
        }
        SessionSet set = sessionSets.get(s.tickTime);
        if (set != null) {
            // 从旧的过期时间的"桶"中移除
            set.sessions.remove(s);
        }
        // 不同的session，它们原始的now + timeout 都是不一样的，
        s.tickTime = expireTime;
        // 经过上面处理后，可能很多session的expireTime是一样的

        set = sessionSets.get(s.tickTime);
        if (set == null) {
            set = new SessionSet();
            sessionSets.put(expireTime, set);
        }
        // 移动到新的过期时间的"桶"中
        set.sessions.add(s);

        // 统一都是expirationInterval的倍数（=tickTime, 2s，在zk配置文件里 zk.cfg）
        // 先检查我的最小倍数（512倍）的那个分桶里的session有没有过期
        // 再检查下一个倍数（618倍）的那个分桶里的session有没有过期

        // expireTime(12:05), SessionSet(<会话1>，<会话2>，<会话3>)
        // expireTime(12:10), SessionSet(<会话4>，<会话5>，<会话6>)
        return true;
    }

    @Override
    synchronized public void setSessionClosing(long sessionId) {
        SessionImpl s = sessionsById.get(sessionId);
        if (s == null) {
            return;
        }
        s.isClosing = true;
    }

    @Override
    synchronized public void removeSession(long sessionId) {
        // sessionsById中移除
        SessionImpl s = sessionsById.remove(sessionId);
        // sessionsWithTimeout中移除
        sessionsWithTimeout.remove(sessionId);
        if (s != null) {
            SessionSet set = sessionSets.get(s.tickTime);
            if(set != null){
                // sessionSets中移除
                set.sessions.remove(s);
            }
        }
    }

    @Override
    public void shutdown() {
        LOG.info("Shutting down");

        running = false;
    }


    @Override
    synchronized public long createSession(int sessionTimeout) {
        // 步骤1：生成一个唯一的id
        // 步骤2：在几个内存数据结构中放入这个session
        // 步骤3：对session计算它的过期时间以及进行特殊的处理
        addSession(nextSessionId, sessionTimeout);
        return nextSessionId++;
    }

    @Override
    synchronized public void addSession(long id, int sessionTimeout) {
        sessionsWithTimeout.put(id, sessionTimeout);
        if (sessionsById.get(id) == null) {
            SessionImpl s = new SessionImpl(id, sessionTimeout, 0);
            sessionsById.put(id, s);
            if (LOG.isTraceEnabled()) {
                ZooTrace.logTraceMessage(LOG, ZooTrace.SESSION_TRACE_MASK, "SessionTrackerImpl --- Adding session 0x" + Long.toHexString(id) + " " + sessionTimeout);
            }
        } else {
            if (LOG.isTraceEnabled()) {
                ZooTrace.logTraceMessage(LOG, ZooTrace.SESSION_TRACE_MASK, "SessionTrackerImpl --- Existing session 0x" + Long.toHexString(id) + " " + sessionTimeout);
            }
        }
        touchSession(id, sessionTimeout);
    }

    @Override
    synchronized public void checkSession(long sessionId, Object owner) throws KeeperException.SessionExpiredException, KeeperException.SessionMovedException {
        SessionImpl session = sessionsById.get(sessionId);
        if (session == null || session.isClosing()) {
            throw new KeeperException.SessionExpiredException();
        }
        if (session.owner == null) {
            session.owner = owner;
        } else if (session.owner != owner) {
            throw new KeeperException.SessionMovedException();
        }
    }

    synchronized public void setOwner(long id, Object owner) throws SessionExpiredException {
        SessionImpl session = sessionsById.get(id);
        if (session == null || session.isClosing()) {
            throw new KeeperException.SessionExpiredException();
        }
        session.owner = owner;
    }
}
